package com.atguigu.gamll.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.gamll.realtime.app.dwd.BaseApp;
import com.atguigu.gamll.realtime.beans.OrderWide;
import com.atguigu.gamll.realtime.beans.PaymentInfo;
import com.atguigu.gamll.realtime.beans.PaymentWide;
import com.atguigu.gamll.realtime.utils.DateUtil;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

// 支付宽表
public class PaymentWideApp extends BaseApp {

    public static void main(String[] args) throws Exception {
        PaymentWideApp wideApp = new PaymentWideApp();
        wideApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {

        // 从kafka中读取数据
        String paymentInfoTopic = "dwd_payment_info";
        String orderWideTopic = "dwm_order_wide";
        String groupId = "paymentInfoGroup";
        FlinkKafkaConsumer<String> paymentInfoSource = MyKafkaUtil.getKafkaSource(paymentInfoTopic, groupId);
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideTopic, groupId);

        DataStreamSource<String> paymentInfoDS = env.addSource(paymentInfoSource);
        DataStreamSource<String> orderWideDS = env.addSource(orderWideSource);

        // 类型转换
        SingleOutputStreamOperator<PaymentInfo> paymentInfoStrDS = paymentInfoDS
                .map(paymentInfoStr -> JSON.parseObject(paymentInfoStr, PaymentInfo.class));
        SingleOutputStreamOperator<OrderWide> orderWideStrDS = orderWideDS
                .map(orderWideStr -> JSON.parseObject(orderWideStr, OrderWide.class));

        // 指定水位线
        SingleOutputStreamOperator<PaymentInfo> paymentInfoWithWaterMarkDS = paymentInfoStrDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<PaymentInfo>() {
                                    @SneakyThrows
                                    @Override
                                    public long extractTimestamp(PaymentInfo paymentInfo, long l) {
                                        return DateUtil.toTs(paymentInfo.getCallback_time());
                                    }
                                }
                        )
        );
        SingleOutputStreamOperator<OrderWide> orderWideWithWaterMarkDS = orderWideStrDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderWide>() {
                                    @SneakyThrows
                                    @Override
                                    public long extractTimestamp(OrderWide orderWide, long l) {
                                        return DateUtil.toTs(orderWide.getCreate_time());
                                    }
                                }
                        )
        );

        // 指定两条流的连接字段
        KeyedStream<PaymentInfo, Long> paymentInfoKeyedDS = paymentInfoWithWaterMarkDS.keyBy(PaymentInfo::getOrder_id);
        KeyedStream<OrderWide, Long> orderWideKeyedDS = orderWideWithWaterMarkDS.keyBy(OrderWide::getOrder_id);

        SingleOutputStreamOperator<PaymentWide> joinedDS = paymentInfoKeyedDS
                .intervalJoin(orderWideKeyedDS)
                .between(Time.seconds(-1800), Time.seconds(0))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo left, OrderWide right, Context ctx, Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(left, right));
                    }
                });

        joinedDS.print(">>>");

        joinedDS
                .map(JSON::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink("dwm_payment_wide"));
    }

}
